#include "transaction.h"
#include <fcntl.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

pthread_mutex_t mute; // 互斥锁

void transaction_sys_init() { pthread_mutex_init(&mute, NULL); }

struct transaction *transaction_new() {
  struct transaction *TX = malloc(sizeof(*TX));
  TX->tables = NULL;
  TX->table_count = 0;

  pthread_mutex_lock(&mute);
  int id = 1;
  if (access("data/transaction", F_OK) != -1) {
    FILE *fp = NULL;
    fp = fopen("data/transaction", "r");
    fscanf(fp, "%d", &id);
    fclose(fp);
    id++;
  }

  FILE *fp = NULL;
  fp = fopen("data/transaction", "w+");
  fprintf(fp, "%d", id);
  fclose(fp);

  TX->id = id;
  pthread_mutex_unlock(&mute);
  
  return TX;
}

// 根据表名查找这是第几张表
int _transaction_find_table(struct transaction *TX, struct string *table_name) {
  for(int i = 0; i<TX->table_count;i++){
      struct table *tbl = TX->tables[i];
      if(string_equals(table_name, tbl->table_name)){
          return i;
      }
  }
  return -1;
}

void _transaction_update_table_pointer(struct transaction *TX, struct table *T) {
  int table_index = _transaction_find_table(TX, T->table_name);
  if (table_index < 0) {
    TX->table_count++;
    if (!TX->tables) {
      TX->tables = malloc(sizeof(struct table *));
    } else {
      TX->tables =
          realloc(TX->tables, sizeof(struct table *) * TX->table_count);
    }
    TX->tables[TX->table_count - 1] = T;
  }
}

int _tx_table_update_row(struct table *T, struct transaction *TX,
                         struct col_name_value_pair **pairs, int pair_count,
                         struct condition **conditions, int condition_count,
                         bool is_insert);

int tx_table_insert_row(struct table *T, struct transaction *TX,
                      struct string **col_vals, bool batch) {
  if (!batch) {
    struct col_name_value_pair **pairs = malloc(sizeof(struct string *) * T->col_count);
    struct condition **conditions = malloc(sizeof(struct string *) * T->col_count);
    for (int i = 0; i < T->col_count; i++) {
      pairs[i] = table_col_name_value_pair_new(string_new(T->col_defs[i]->col_name->chs), string_new(col_vals[i]->chs));
      conditions[i] = condition_new(string_new(T->col_defs[i]->col_name->chs), EQUALS, string_new(""));
    }
    int res = _tx_table_update_row(T, TX, pairs, T->col_count, conditions,
                                T->col_count, true);
    for (int i = 0; i < T->col_count; i++) {
      table_col_name_value_pair_free(pairs[i]);
      condition_free(conditions[i]);
    }
    free(pairs);
    free(conditions);
    if (res == 1) {
      return res;
    }
  } else {
    _transaction_update_table_pointer(TX, T);
  }

  FILE *tx_fp = NULL;
  char tx_file_name[60];
  sprintf(tx_file_name, "data/tx_%d_%s", TX->id, T->table_name->chs);
  tx_fp = fopen(tx_file_name, "a+");

  for (int i = 0; i < T->col_count; i++) {
    fputs(col_vals[i]->chs, tx_fp);
    for (int j = 0; j < T->col_defs[i]->length - col_vals[i]->len; j++) {
      fputc(' ', tx_fp);
    }
  }
  // 数据原始行号填充成空格
  for (int i = 0; i < 10; i++) {
    fputc(' ', tx_fp);
  }
  fclose(tx_fp);

  return 1;
}

int tx_table_update_row(struct table *T, struct transaction *TX,
                        struct col_name_value_pair **pairs, int pair_count,
                        struct condition **conditions, int condition_count) {
  return _tx_table_update_row(T, TX, pairs, pair_count, conditions,
                              condition_count, false);
}

int _tx_table_update_row(struct table *T, struct transaction *TX,
                         struct col_name_value_pair **pairs, int pair_count,
                         struct condition **conditions, int condition_count,
                         bool is_insert) {
  // TX内的表指针更新
  _transaction_update_table_pointer(TX, T);
  // 新建文件/打开文件
  FILE *fp = NULL;
  char file_name[50] = "data/";
  strcat(file_name, T->table_name->chs);
  if (access(file_name, F_OK) == -1) {
    fp = fopen(file_name, "a+");
    fclose(fp);
  }
  fp = fopen(file_name, "r+");

  FILE *tx_fp = NULL;
  char tx_file_name[60];
  sprintf(tx_file_name, "data/tx_%d_%s", TX->id, T->table_name->chs);
  if (access(tx_file_name, F_OK) == -1) {
    tx_fp = fopen(tx_file_name, "a+");
    fclose(tx_fp);
  }
  tx_fp = fopen(tx_file_name, "r+");

  int update_count = 0;

  // 对tx_$id_$table_name文件当中已有的符合条件的数据进行更新
  struct stat tx_statbuf;
  stat(tx_file_name, &tx_statbuf);
  long long tx_row_count = tx_statbuf.st_size / (T->row_length + 10);

  for(int i = 0; i < tx_row_count; i++){
    // 如果是空行且不是插入操作，可以忽略这一行
    if (_table_row_is_empty(T, tx_fp, i) && !is_insert) {
      continue;
    }

    // 如果跟查询条件不匹配，可以忽略这一行
    if (!_table_row_match_condition(T, conditions, condition_count, tx_fp, i)) {
      continue;
    }

    update_count++;
    for (int j = 0; j < T->col_count; j++) {
      fseek(tx_fp, (T->row_length + 10) * (long long)i + _table_get_col_offset(T, j), 0);
      struct col_def *col_def = T->col_defs[j];
      struct col_name_value_pair *pair =
          table_find_col_name_value_pair(pairs, pair_count, col_def->col_name);
      if (pair) {
        fputs(pair->col_value->chs, tx_fp);
        for (int k = 0; k < col_def->length - pair->col_value->len; k++) {
          fputc(' ', tx_fp);
        }
      }
    }
    if (is_insert && update_count == 1) {
      fclose(fp);
      fclose(tx_fp);
      return update_count;
    }
  }

  pthread_rwlock_rdlock(&T->rwlock_for_data);
  struct stat statbuf;
  stat(file_name, &statbuf);
  long long row_count = statbuf.st_size / (T->row_length + 10); // 怎么样知道数据文件行数目前为止只想到这个笨办法
  pthread_rwlock_unlock(&T->rwlock_for_data);

  // 全表扫描并更新数据到tx_$id_$table_name文件当中去
  // 对符合条件的数据加行锁，并将数据更新写入到tx_$id_$table_name文件当中
  for (int i = 0; i < row_count; i++) {
    pthread_rwlock_rdlock(&T->rwlock_for_data);

    // 如果是空行且不是插入操作，可以忽略这一行
    if (_table_row_is_empty(T, fp, i) && !is_insert) {
      pthread_rwlock_unlock(&T->rwlock_for_data);
      continue;
    }

    // 如果跟查询条件不匹配，不需要关心这一行是否被当前事务修改过，可以忽略这一行
    if (!_table_row_match_condition(T, conditions, condition_count, fp, i)) {
      pthread_rwlock_unlock(&T->rwlock_for_data);
      continue;
    }

    pthread_rwlock_rdlock(&T->rwlock_for_tx);
    fseek(fp, (T->row_length + 10) * (long long)i + T->row_length, 0);
    struct string *value = _table_read_value_from_file(fp, 10);
    pthread_rwlock_unlock(&T->rwlock_for_tx);
    // 如果跟查询条件匹配且行锁已存在且事务id与当前事务一致，说明已经由前面的代码判断过了，可以忽略这一行
    if (value->len > 0 && atoi(value->chs) == TX->id) {
      string_free(value);
      pthread_rwlock_unlock(&T->rwlock_for_data);
      continue;
    } else if (value->len > 0) {
      // 如果跟查询条件匹配且行锁已存在且事务id与当前事务不一致，说明其他事务正在修改这一行，隔100ms后重试
      string_free(value);
      pthread_rwlock_unlock(&T->rwlock_for_data);
      while (true) {
        usleep(100000);
        pthread_rwlock_rdlock(&T->rwlock_for_tx);
        fclose(fp); // 数据文件中已经释放行锁了，可这边读到的还是旧的行锁，先用重新打开文件的方法解决下
        fp = fopen(file_name, "r+");
        fseek(fp, (T->row_length + 10) * (long long)i + T->row_length, 0);
        value = _table_read_value_from_file(fp, 10);
        pthread_rwlock_unlock(&T->rwlock_for_tx);
        if (value->len == 0) {
          string_free(value);
          break;
        }
        string_free(value);
      }
      pthread_rwlock_rdlock(&T->rwlock_for_data);
      // 如果是空行且不是插入操作，可以忽略这一行
      if (_table_row_is_empty(T, fp, i) && !is_insert) {
        pthread_rwlock_unlock(&T->rwlock_for_data);
        continue;
      }

      // 如果跟查询条件不匹配，不需要关心这一行是否被当前事务修改过，可以忽略这一行
      if (!_table_row_match_condition(T, conditions, condition_count, fp, i)) {
        pthread_rwlock_unlock(&T->rwlock_for_data);
        continue;
      }
    } else {
      string_free(value);
    }

    // 如果跟查询条件匹配且行锁不存在，先获取mute互斥锁，再次判断行锁是否存在：
    pthread_rwlock_wrlock(&T->rwlock_for_tx);
    // 如果再次校验发现这一行的事务id位置上有内容了，释放mute互斥锁并抛异常还是一直占着mute互斥锁每隔100ms重试一次？
    fseek(fp, (T->row_length + 10) * (long long)i + T->row_length, 0);
    value = _table_read_value_from_file(fp, 10);
    if (value->len > 0) {
      string_free(value);
      pthread_rwlock_unlock(&T->rwlock_for_tx);
      pthread_rwlock_unlock(&T->rwlock_for_data);
      fclose(fp);
      fclose(tx_fp);
      return -1;
    }
    // 如果不存在，说明没有其他线程/事务在操作这一行，给这一行的事务id位置写上自己的事务id，然后释放mute互斥锁
    fseek(fp, (T->row_length + 10) * (long long)i + T->row_length, 0);
    fprintf(fp, "%d", TX->id);
    pthread_rwlock_unlock(&T->rwlock_for_tx);

    update_count++;
    for (int j = 0; j < T->col_count; j++) {
      fseek(fp, (T->row_length + 10) * (long long)i + _table_get_col_offset(T, j), 0);
      struct col_def *col_def = T->col_defs[j];
      struct string *value = _table_read_value_from_file(fp, col_def->length);
      struct col_name_value_pair *pair =
          table_find_col_name_value_pair(pairs, pair_count, col_def->col_name);
      if (pair) {
        string_free(value);
        value = string_new(pair->col_value->chs);
      }
      fputs(value->chs, tx_fp);
      for (int k = 0; k < col_def->length - value->len; k++) {
        fputc(' ', tx_fp);
      }
      string_free(value);
    }
    struct string *row_index_str = string_value_of(i);
    fputs(row_index_str->chs, tx_fp);
    for (int j = 0; j < 10 - row_index_str->len; j++) {
      fputc(' ', tx_fp);
    }
    string_free(row_index_str);

    pthread_rwlock_unlock(&T->rwlock_for_data);

    if (is_insert && update_count == 1) {
      fclose(fp);
      fclose(tx_fp);
      return update_count;
    }
  }

  fclose(fp);
  fclose(tx_fp);
  
  return update_count;
}

int tx_table_delete_row(struct table *T, struct transaction *TX,
                        struct condition **conditions, int condition_count) {
  struct col_name_value_pair **pairs = malloc(sizeof(struct string *) * T->col_count);
  for (int i = 0; i < T->col_count; i++) {
    pairs[i] = table_col_name_value_pair_new(string_new(T->col_defs[i]->col_name->chs), string_new(""));
  }
  int res = tx_table_update_row(T, TX, pairs, T->col_count, conditions, condition_count);
  for (int i = 0; i < T->col_count; i++) {
    table_col_name_value_pair_free(pairs[i]);
  }
  free(pairs);
  return res;
}

struct string *tx_table_select(struct table *T, struct transaction *TX,
                               struct condition **conditions,
                               int condition_count) {
  struct string *result = string_new("");
  // 拼接列名到结果字符串
  for (int i = 0; i < T->col_count; i++) {
    string_append(result, T->col_defs[i]->col_name->chs);
    string_append(result, "\t");
  }
  string_append(result, "\n");

  // 从tx_$id_$table_name文件当中查询数据
  FILE *tx_fp = NULL;
  char tx_file_name[60];
  sprintf(tx_file_name, "data/tx_%d_%s", TX->id, T->table_name->chs);
  if (access(tx_file_name, F_OK) != -1) {
    tx_fp = fopen(tx_file_name, "r+");

    struct stat tx_statbuf;
    stat(tx_file_name, &tx_statbuf);
    long long tx_row_count = tx_statbuf.st_size / (T->row_length + 10);

    for (int i = 0; i < tx_row_count; i++) {

      // 如果是空行，可以忽略这一行
      if(_table_row_is_empty(T, tx_fp, i)){
        continue;
      }

      // 如果跟查询条件不匹配，可以忽略这一行
      if (!_table_row_match_condition(T, conditions, condition_count, tx_fp, i)) {
        continue;
      }

      for (int j = 0; j < T->col_count; j++) {
        fseek(tx_fp, (T->row_length + 10) * (long long)i + _table_get_col_offset(T, j), 0);
        struct col_def *col_def = T->col_defs[j];
        struct string *value = _table_read_value_from_file(tx_fp, col_def->length);
        string_append(result, value->chs);
        string_append(result, "\t");
        string_free(value);
      }
      string_append(result, "\n");
    }
    fclose(tx_fp);
  }

  FILE *fp = NULL;
  char file_name[50] = "data/";
  strcat(file_name, T->table_name->chs);
  if (access(file_name, F_OK) == -1) {
    fp = fopen(file_name, "a+");
    fclose(fp);
  }
  fp = fopen(file_name, "r");

  pthread_rwlock_rdlock(&T->rwlock_for_data);
  struct stat statbuf;
  stat(file_name, &statbuf);
  long long row_count = statbuf.st_size / (T->row_length + 10);  // 怎么样知道数据文件行数目前为止只想到这个笨办法
  pthread_rwlock_unlock(&T->rwlock_for_data);

  // 拼接数据到结果字符串
  for (int i = 0; i < row_count; i++) {

    pthread_rwlock_rdlock(&T->rwlock_for_data);

    // 如果是空行，可以忽略这一行
    if (_table_row_is_empty(T, fp, i)) {
      pthread_rwlock_unlock(&T->rwlock_for_data);
      continue;
    }

    if (!_table_row_match_condition(T, conditions, condition_count, fp, i)) {
      pthread_rwlock_unlock(&T->rwlock_for_data);
      continue;
    }

    //  如果跟查询条件匹配且行锁已存在且事务id与当前事务一致，说明已经由前面的代码判断过并拼接到结果字符串了，可以忽略这一行
    pthread_rwlock_rdlock(&T->rwlock_for_tx);
    fseek(fp, (T->row_length + 10) * (long long)i + T->row_length, 0);
    struct string *value = _table_read_value_from_file(fp, 10);
    pthread_rwlock_unlock(&T->rwlock_for_tx);
    if (value->len > 0 && atoi(value->chs) == TX->id) {
      string_free(value);
      pthread_rwlock_unlock(&T->rwlock_for_data);
      continue;
    }

    for (int j = 0; j < T->col_count; j++) {
      fseek(fp, (T->row_length + 10) * (long long)i + _table_get_col_offset(T, j), 0);
      struct col_def *col_def = T->col_defs[j];
      struct string *value = _table_read_value_from_file(fp, col_def->length);
      string_append(result, value->chs);
      string_append(result, "\t");
      string_free(value);
    }
    string_append(result, "\n");
    pthread_rwlock_unlock(&T->rwlock_for_data);
  }

  fclose(fp);
  return result;
}

void _transaction_commit_or_rollback(struct transaction *TX, bool is_commit) {
  for (int i = 0; i < TX->table_count; i++) {
    struct table *T = TX->tables[i];

    FILE *tx_fp = NULL;
    char tx_file_name[60];
    sprintf(tx_file_name, "data/tx_%d_%s", TX->id, T->table_name->chs);
    if (access(tx_file_name, F_OK) == -1) {
      continue;
    }
    tx_fp = fopen(tx_file_name, "r+");

    FILE *fp = NULL;
    char file_name[50] = "data/";
    strcat(file_name, T->table_name->chs);
    if (access(file_name, F_OK) == -1) {
      fp = fopen(file_name, "a+");
      fclose(fp);
    }
    fp = fopen(file_name, "r+");

    pthread_rwlock_wrlock(&T->rwlock_for_data);
    struct stat tx_statbuf;
    stat(tx_file_name, &tx_statbuf);
    long long tx_row_count = tx_statbuf.st_size / (T->row_length + 10);

    for (int i = 0; i < tx_row_count; i++) {
      fseek(tx_fp, (T->row_length + 10) * (long long)i + T->row_length, 0);
      struct string *value = _table_read_value_from_file(tx_fp, 10);
      if (is_commit) {
        if (value->len == 0) {
          fseek(fp, 0, 2);
        } else {
          int row_index = atoi(value->chs);
          fseek(fp, (T->row_length + 10) * (long long)row_index, 0);
        }
        string_free(value);
        fseek(tx_fp, (T->row_length + 10) * (long long)i, 0);
        char buffer[T->row_length];
        for (int i = 0; i < T->row_length; i++) {
          buffer[i] = fgetc(tx_fp);
        }
        for (int i = 0; i < T->row_length; i++) {
          fputc(buffer[i], fp);
        }
        pthread_rwlock_wrlock(&T->rwlock_for_tx);
        for (int i = 0; i < 10; i++) {
          fputc(' ', fp);
        }
        pthread_rwlock_unlock(&T->rwlock_for_tx);
      } else {
        if (value->len > 0) {
          int row_index = atoi(value->chs);
          pthread_rwlock_wrlock(&T->rwlock_for_tx);
          fseek(fp, (T->row_length + 10) * (long long)row_index + T->row_length, 0);
          for (int i = 0; i < 10; i++) {
            fputc(' ', fp);
          }
          pthread_rwlock_unlock(&T->rwlock_for_tx);
        }
        string_free(value);
      }
    }
    pthread_rwlock_unlock(&T->rwlock_for_data);
    fclose(fp);
    fclose(tx_fp);
    // 事务提交完以后清空一下tx_$id_$table_name文件
    tx_fp = fopen(tx_file_name, "w+");
    fclose(tx_fp);
  }
}

void transaction_commit(struct transaction *TX) {
  _transaction_commit_or_rollback(TX, true);
}

void transaction_rollback(struct transaction *TX) {
  _transaction_commit_or_rollback(TX, false);
}

void transaction_free(struct transaction *TX) {

  // 删除tx_$id_$table_name文件
  for (int i = 0; i < TX->table_count; i++) {
    struct table *T = TX->tables[i];
    char tx_file_name[60];
    sprintf(tx_file_name, "data/tx_%d_%s", TX->id, T->table_name->chs);
    if (access(tx_file_name, F_OK) != -1) {
      remove(tx_file_name);
    }
  }
  free(TX->tables);
  free(TX);
}